package com.leyantech.ecp.agenx.dispatcher.sender;

import com.leyantech.ecp.agenx.common.config.Constants;
import com.leyantech.ecp.agenx.common.sender.Sender;

import com.google.inject.Inject;
import com.google.inject.Singleton;
import io.opentracing.Span;
import io.opentracing.util.GlobalTracer;
import io.vertx.core.Future;
import io.vertx.core.eventbus.impl.EventBusImpl;
import io.vertx.core.eventbus.impl.MessageImpl;
import io.vertx.core.impl.VertxInternal;
import io.vertx.tracing.opentracing.OpenTracingUtil;

/**
 * @author eahau, {@literal <eahau@foxmail.com>}
 * @date 2022-11-24.
 */
@Singleton
public class EventBusSender implements Sender<String> {

  private final VertxInternal vertx;

  private final EventBusImpl eventBus;

  @Inject
  public EventBusSender(VertxInternal vertx) {
    this.vertx = vertx;
    this.eventBus = (EventBusImpl) vertx.eventBus();
  }

  @Override
  public Future<Void> send(String address, String message) {
    final Span span = GlobalTracer.get().activeSpan();
    return vertx.executeBlocking(promise -> {
      OpenTracingUtil.setSpan(span);
      MessageImpl msg = eventBus.createMessage(false, address, null, message, null);
      eventBus.sendOrPubInternal(msg, Constants.DEFAULT_DELIVERY_OPTIONS, null, promise);
    });
  }

}
